#include "testcommon.h"
#include "zeppreactor.h"
#include <hiredis/alloc.h>
#include <hiredis/async.h>
#include <hiredis/hiredis.h>

struct redisReactorEvent {
  struct zeppevent ev;
  redisAsyncContext *ctx;
  struct zeppreactor *reactor;
};

int redisReadHandel(int fd, uint32_t event, void *arg) {
  (void)fd;
  (void)event;
  struct redisReactorEvent *r = (struct redisReactorEvent *)arg;
  redisAsyncHandleRead(r->ctx);
  return 0;
}

int redisWriteHandel(int fd, uint32_t event, void *arg) {
  (void)fd;
  (void)event;
  struct redisReactorEvent *r = (struct redisReactorEvent *)arg;
  redisAsyncHandleWrite(r->ctx);
  return 0;
}

void redisAddRead(void *privdata) {
  // printf("redisAddRead\n");
  struct redisReactorEvent *r = (struct redisReactorEvent *)privdata;
  zeppevent_set(&r->ev, r->ev.fd, redisReadHandel, privdata);

  zeppevent_del(r->reactor->epfd, &r->ev);
  zeppevent_add(r->reactor->epfd, EPOLLIN, &r->ev);
}

void redisDelRead(void *privdata) {
  // printf("redisDelRead\n");
  struct redisReactorEvent *r = (struct redisReactorEvent *)privdata;
  zeppevent_set(&r->ev, r->ev.fd, redisReadHandel, privdata);

  zeppevent_del(r->reactor->epfd, &r->ev);
  zeppevent_add(r->reactor->epfd, 0, &r->ev);
}

void redisAddWrite(void *privdata) {
  // printf("redisAddWrite\n");
  struct redisReactorEvent *r = (struct redisReactorEvent *)privdata;
  zeppevent_set(&r->ev, r->ev.fd, redisWriteHandel, privdata);

  zeppevent_del(r->reactor->epfd, &r->ev);
  zeppevent_add(r->reactor->epfd, EPOLLOUT, &r->ev);
}

void redisDelWrite(void *privdata) {
  // printf("redisDelWrite\n");
  struct redisReactorEvent *r = (struct redisReactorEvent *)privdata;
  zeppevent_set(&r->ev, r->ev.fd, redisWriteHandel, privdata);

  zeppevent_del(r->reactor->epfd, &r->ev);
  zeppevent_add(r->reactor->epfd, 0, &r->ev);
}

void redisClearUP(void *privdata) {
  // printf("redisClearUP\n");
  struct redisReactorEvent *r = (struct redisReactorEvent *)privdata;
  struct zeppreactor *reactor = r->reactor;
  zeppevent_del(reactor->epfd, &r->ev);
  hi_free(r);
}

static int redisReactorAttach(redisAsyncContext *ac,
                              struct zeppreactor *reactor) {
  redisContext *c = &ac->c;
  struct redisReactorEvent *re;

  re = (struct redisReactorEvent *)hi_malloc(sizeof(*re));
  if (re == NULL)
    return REDIS_ERR;

  re->ctx = ac;
  // use hiredis buffer;
  re->ev.readbuffer = NULL;
  re->ev.writebuffer = NULL;
  re->ev.fd = c->fd;
  re->reactor = reactor;

  ac->ev.addRead = redisAddRead;
  ac->ev.delRead = redisDelRead;
  ac->ev.addWrite = redisAddWrite;
  ac->ev.delWrite = redisDelWrite;
  ac->ev.cleanup = redisClearUP;
  ac->ev.data = re;

  return REDIS_OK;
}

// 获取当前时间的毫秒数
int current_tick() {
  int t = 0;
  struct timespec ti;
  clock_gettime(CLOCK_MONOTONIC, &ti);
  t = (int)ti.tv_sec * 1000;
  t += ti.tv_nsec / 1000000;
  return t;
}
static int cnt, before, num;
static struct zeppreactor reactor;

void connectCallback(const redisAsyncContext *c, int status) {
  if (status != REDIS_OK) {
    printf("Error: %s\n", c->errstr);
    return;
  }
  printf("Connected...\n");
}

void disconnectCallback(const redisAsyncContext *c, int status) {
  if (status != REDIS_OK) {
    printf("Error: %s\n", c->errstr);
    return;
  }
  printf("Disconnected...\n");
  zeppreactor_destory(&reactor);
}

void getCallback(redisAsyncContext *ac, void *r, void *privdata) {
  redisReply *reply = r;
  if (reply == NULL)
    return;
  // printf("argv[%s]: %lld\n", (char *)privdata, reply->integer);

  cnt++;
}

void getRetCallback(redisAsyncContext *ac, void *r, void *privdata) {
  redisReply *reply = r;
  printf("counter: %s\n", reply->str);
  int used = current_tick() - before;
  printf("After %d exec redis commands, used %d ms\n", num, used);
  redisAsyncDisconnect(ac);
}

TEST_SUITE(async_redis) {
  zeppreactor_init(&reactor);

  redisAsyncContext *ac = redisAsyncConnect("127.0.0.1", 6379);
  if (ac->err) {
    printf("Error: %s\n", ac->errstr);
    return;
  }

  redisReactorAttach(ac, &reactor);

  redisAsyncSetConnectCallback(ac, connectCallback);
  redisAsyncSetDisconnectCallback(ac, disconnectCallback);

  num = 10;

  before = current_tick();
  for (int i = 0; i < num; i++)
    redisAsyncCommand(ac, getCallback, "count", "INCR counter");
  redisAsyncCommand(ac, getRetCallback, NULL, "get counter");

  zeppreactor_run(&reactor);
}

TEST_SUITE(sync_redis) {
  redisContext *c = redisConnect("127.0.0.1", 6379);
  if (c->err) {
    printf("Error: %s\n", c->errstr);
    return;
  }
  num = 1000;
  before = current_tick();
  for (int i = 0; i < num; i++) {
    redisReply *reply = redisCommand(c, "INCR counter");
    // printf("argv[count]: %lld\n", reply->integer);
    freeReplyObject(reply);
  }
  redisReply *reply = redisCommand(c, "get counter");
  printf("counter: %s\n", reply->str);
  freeReplyObject(reply);
  int used = current_tick() - before;
  printf("After %d exec redis commands, used %d ms\n", num, used);

  redisFree(c);
}

int main() {
  RUN_TEST_SUITE(async_redis);
  RUN_TEST_SUITE(sync_redis);
  return 0;
}